/*
 * line protocol 到 MQTT格式的转换器，将一组以换行符分割的报文转换为json字符串
 * 首先，通过mapToEquip方法将数据分组
 * 其次，针对分组后的数据一个设备组装一个RootCloudThing对象
 * 最后，通过 gson 将封装后的map转换成 MQTT 可以接受的格式
 * payload格式如下：
 * {
    "body": {
        "things": [{
            "id": "device_1590557277",
            "thingType": "Device",
            "items": [{
                "qBad": ["level"],
                "ts": 1590546059547,
                "properties": {
                    "temp": 13,
                    "level": 75
                }
            }]
        }, {
            "id": "device_1590557589",
            "thingType": "Device",
            "items": [{
                "qBad": [],
                "ts": 1590546059547,
                "properties": {
                    "temp": 18,
                    "__raw_loc__":  {"gps":"$GPRMC,092927.000,A,2235.9058,N,11400.0518,E,0.000,74.11,151216,,,D*49"}
                }
            }]
        }]
    }
}
*/
package com.luis.toolbox.handler.rc;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import com.luis.toolbox.common.EquipMappingConfig;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import com.google.gson.Gson;
import com.luis.toolbox.common.Constants;
import com.luis.toolbox.common.Constants.ThingType;
import com.luis.toolbox.handler.rc.model.RootCloudItem;
import com.luis.toolbox.handler.rc.model.RootCloudThing;
import com.luis.toolbox.intf.MyTransformer;

@Component
@ConditionalOnProperty(prefix="outbound",name = "destination", havingValue = "rc")
public class RootCloudTransformer implements MyTransformer{
	
	private final Logger log = LoggerFactory.getLogger(MyTransformer.class);

	@Value("${sample-count:-1}")
	private int sampleCount;
	@Autowired
	private EquipMappingConfig mapping;

	@Override
	public String transform(String lineData) {
		if(sampleCount == 0) return "";
		Map<String,List<String>> equipDataMap = Constants.mapToEquip(lineData);
		List<RootCloudThing> thingsList = new ArrayList<>();
		AtomicInteger dataTotal = new AtomicInteger();
		equipDataMap.forEach((k,v)->{
			//通过设备获取通讯标识，如果没有配置通讯标识的设备则被过滤掉
			String target = mapping.get(k);
			if(StringUtils.isBlank(target)) return;
			RootCloudThing thing = new RootCloudThing(target,ThingType.DEVICE);
			List<RootCloudItem> items = v.stream()
					.map(RootCloudItem::buildFromLine).collect(Collectors.toList());
			//将items以时间维度进行聚合
			List<RootCloudItem> targetItem = groupItemByTime(items);
			//对聚合结果进行取样, >0进行取样，=0 不发送， <0 全量发送,
			if(sampleCount >0){
				List<RootCloudItem> sampleList = Constants.doSampling(targetItem, sampleCount);
				thing.setItems(sampleList);
				dataTotal.addAndGet(sampleList.size());
			}else{
				thing.setItems(targetItem);
				dataTotal.addAndGet(items.size());
			}

			thingsList.add(thing);
		});
		//如果没有数据则不发送
		if(dataTotal.get() == 0) return "";

		Map<String,List<RootCloudThing>> innerMap = new HashMap<>();
		innerMap.put("things", thingsList);
		Map<String,Object> outerMap = new HashMap<>();
		outerMap.put("body", innerMap);
		log.info("success to tranform {} message for {} equipment[s]", dataTotal.get(), thingsList.size());
		return new Gson().toJson(outerMap);
	}

	private List<RootCloudItem> groupItemByTime(List<RootCloudItem> sourceList){
		Map<Long, Optional<RootCloudItem>> result = sourceList.stream().collect(Collectors
				.groupingBy(RootCloudItem::getTs,Collectors.reducing(
						(a,b)->{
							a.getProperties().putAll(b.getProperties());
							return a;
						})));
		return result.values().stream().filter(Optional::isPresent)
				.map(Optional::get)
				.sorted(Comparator.comparingLong(RootCloudItem::getTs)).collect(Collectors.toList());
	}
}
